package com.desheng.bigdata.flink.stream.transformation

import org.apache.flink.streaming.api.functions.ProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.util.Collector

/**
  * flink算子操作之split和select
  *    split是将一个datasteam根据指定的split函数拆分成为一个datastream的集--->splitstream
  *    select就是从上述拆分的流splitstream中去选择其中某一个或多个流
  *
  * 自flink1.8.2开始，split和select api便不再被建议使用，转而建议使用侧输出流进行替代
  *
  */
object _02TransformationForSideOutputOps {
    def main(args: Array[String]): Unit = {
        val env = StreamExecutionEnvironment.getExecutionEnvironment
        env.setParallelism(1)

        val inputs:DataStream[String] = env.socketTextStream("bigdata01", 9999)

        val goods = inputs.map(line => {
            val fields = line.split("\\|")
            if(fields == null || fields.length != 3) {
                Goods(null, null, null)
            } else {
                Goods(fields(0), fields(1), fields(2))
            }
        }).filter(goods => goods.id != null)
        //通过process函数来完成流的侧输出,其实就是对流中的每一条记录打上一个标签
        //提前定义标签
        val mobileTag =  new OutputTag[Goods]("mobile")
        val sportsTag = new OutputTag[Goods]("sports")
        val clothingTag = new OutputTag[Goods]("clothing")
        val tagStream: DataStream[Goods] = goods.process(new ProcessFunction[Goods, Goods] {
            override def processElement(goods: Goods,
                                        ctx: ProcessFunction[Goods, Goods]#Context,
                                        out: Collector[Goods]): Unit = {
                goods.category match {
                    case "mobile" => {
                        ctx.output(mobileTag, goods)
                    }
                    case "sports" => {
                        ctx.output(sportsTag, goods)
                    }
                    case "clothing" => {
                        ctx.output(clothingTag, goods)
                    }
                    case _ => {
                        out.collect(goods)
                    }
                }
            }
        })

        //获取的时候，就可以根据指定的tag来获取相关的标签数据

        tagStream.getSideOutput(clothingTag).print("clothing")

        env.execute(s"${_01TransformationForSplitAndSelectOps.getClass.getSimpleName}")
    }
}
